package com.pace.cloud.common.entities.kafka.util;

import com.alibaba.fastjson.JSONObject;
import com.sun.istack.internal.NotNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;

import java.util.Map;

@Slf4j
public class MessageUtils {

    public final static String TOPIC_MESSAGE = "test";

    private static KafkaTemplate TEMPLATE;

    private MessageUtils() {
    }

    public static void init(KafkaTemplate kafkaTemplate) {
        TEMPLATE = kafkaTemplate;
    }


    /**
     * @param messageTypeCode
     * @param paramMap
     */
    public static void addMessage(@NotNull String messageTypeCode, Map<String, Object> paramMap) {
        JSONObject data = new JSONObject();
        data.put("paramMap", paramMap);
        send(TOPIC_MESSAGE, messageTypeCode, data.toJSONString());
    }

    /**
     * @param topic
     * @param type
     * @param jsonData
     */
    private static void send(String topic, String type, String jsonData) {
        if (StringUtils.isEmpty(TEMPLATE)) {
            log.warn("Kafka is not enabled, please check config.");
            return;
        }

        log.info("send data to kafka, topic:{}, type:{}, jsonData: {}", topic, type, jsonData);
        TEMPLATE.send(topic, type, jsonData);
    }
}
